package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.ProvinceOrderInfo;
import com.atguigu.edu.bean.ProvinceTransactionOrderInfo;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsProvinceTransactionWindow {
    public static void main(String[] args) throws Exception {
        //todo 1 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //todo 2 设置状态后端
        /*env.enableCheckpointing(5 * 1000L, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 *1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");
        env.setStateBackend(new HashMapStateBackend());*/

        //todo 3 从kafka获取主题数据
        String topicName = "dwd_trade_order_detail";
        String groupId = "dws_province_transaction_window";
        DataStreamSource<String> dataStreamSource = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        //todo 4 转换数据类型
        SingleOutputStreamOperator<ProvinceTransactionOrderInfo> mapStream = dataStreamSource.map(new MapFunction<String, ProvinceTransactionOrderInfo>() {
            @Override
            public ProvinceTransactionOrderInfo map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                ProvinceTransactionOrderInfo orderInfo = new ProvinceTransactionOrderInfo();
                orderInfo.setProvinceId(jsonObject.getString("province_id"));
                orderInfo.setUserId(jsonObject.getString("user_id"));
                orderInfo.setOriginalAmount(jsonObject.getDouble("origin_amount"));
                orderInfo.setCouponAmount(jsonObject.getDouble("coupon_reduce"));
                orderInfo.setOrderAmount(jsonObject.getDouble("origin_amount") - jsonObject.getDouble("coupon_reduce"));
                orderInfo.setTs(jsonObject.getLong("od_ts"));
                return orderInfo;
            }
        });

        //todo 5 根据省份id分组 并根据状态去重userId
        KeyedStream<ProvinceTransactionOrderInfo, String> keyByStream = mapStream
                .keyBy(ProvinceTransactionOrderInfo -> ProvinceTransactionOrderInfo.getProvinceId());

        SingleOutputStreamOperator<ProvinceOrderInfo> userIdStream = keyByStream.flatMap(new RichFlatMapFunction<ProvinceTransactionOrderInfo, ProvinceOrderInfo>() {
            private ValueState<String> userVisitDt = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                userVisitDt = getRuntimeContext()
                        .getState(new ValueStateDescriptor<String>("user_visit_dt", String.class));
            }

            @Override
            public void flatMap(ProvinceTransactionOrderInfo provinceTransactionOrderInfo, Collector<ProvinceOrderInfo> collector) throws Exception {
                String userId = provinceTransactionOrderInfo.getUserId();
                ProvinceOrderInfo provinceOrderInfo = new ProvinceOrderInfo();
                provinceOrderInfo.setOrderAmount(provinceTransactionOrderInfo.getOrderAmount());
                provinceOrderInfo.setProvinceId(provinceTransactionOrderInfo.getProvinceId());
                provinceOrderInfo.setOrderCount(1L);
                //provinceOrderInfo.setOrderAmount(provinceTransactionOrderInfo.getOrderAmount());
                provinceOrderInfo.setTs(provinceTransactionOrderInfo.getTs());
                if (userVisitDt == null || !(userId).equals(userVisitDt.value())) {
                    userVisitDt.update(userId);
                    provinceOrderInfo.setUserCount(1L);
                }else{
                    provinceOrderInfo.setUserCount(0L);
                }
                collector.collect(provinceOrderInfo);
            }
        });
        //todo 6 设置水位线
        SingleOutputStreamOperator<ProvinceOrderInfo> watermarkStream = userIdStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<ProvinceOrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner(new SerializableTimestampAssigner<ProvinceOrderInfo>() {
                    @Override
                    public long extractTimestamp(ProvinceOrderInfo element, long recordTimestamp) {
                        return element.getTs() * 1000L;
                    }
                }));
        //todo 7 开窗聚合
        SingleOutputStreamOperator<ProvinceOrderInfo> reduceStream = watermarkStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10L))).reduce(new ReduceFunction<ProvinceOrderInfo>() {
            @Override
            public ProvinceOrderInfo reduce(ProvinceOrderInfo value1, ProvinceOrderInfo value2) throws Exception {
                value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                value1.setUserCount(value1.getUserCount() + value2.getUserCount());
                return value1;
            }
        }, new AllWindowFunction<ProvinceOrderInfo, ProvinceOrderInfo, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<ProvinceOrderInfo> values, Collector<ProvinceOrderInfo> out) throws Exception {
                ProvinceOrderInfo next = values.iterator().next();
                next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                next.setTs(System.currentTimeMillis());
                out.collect(next);
            }
        });
        reduceStream.print();

        //todo 8 写入clickHouse中
        reduceStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_province_transaction_window values (?,?,?,?,?,?,?)"));

        //todo 9 执行任务
        env.execute(groupId);
    }

}
